package com.bw.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.bean.TrafficPageViewBean;
import com.bw.gmall.realtime.utils.MyClickHouseUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;

/*
*
* dwd 独立访客
* dwd 跳出
* dwd 分流
* */
public class DwsTrafficVcChArIsNewPageViewWindow {

    public static void main(String[] args) throws Exception {

        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //TODO 2.读取三个主题的数据创建流
        String uvTopic = "dwd_traffic_unique_visitor_detail_project";//          独立访客
        String ujdTopic = "dwd_traffic_user_jump_detail_project"; //             跳出用户数
        String topic = "dwd_traffic_page_log_project_shop";  //执行时间   流量量       会话数、页面浏览数和浏览总时长
        String groupId = "vccharisnew_pageview_window_2103a";

       //json  {}
        //创建对象    vc   ch  ar  is_new  uv   jv  pv   sv duringtime
        //  vc   ch  ar  is_new  uv =1  jv=0  pv=0   sv=0  duringtime=0
        //  vc   ch  ar  is_new  uv =0  jv=1  pv=0   sv=0  duringtime=0  时效性收运行
        //  vc   ch  ar  is_new  uv =0  jv=0  pv=1   sv=1 只要上一是null  duringtime=during_time


        //uv
        DataStreamSource<String> uvDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(uvTopic, groupId));

        DataStreamSource<String> ujDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(ujdTopic, groupId));

        DataStreamSource<String> pageDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

        SingleOutputStreamOperator<TrafficPageViewBean> uvDS1 = (SingleOutputStreamOperator<TrafficPageViewBean>) uvDS.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject common = jsonObject.getJSONObject("common");
                JSONObject page = jsonObject.getJSONObject("page");
                return new TrafficPageViewBean(null,
                        null,
                        common.getString("vc"),
                        common.getString("ch"),
                        common.getString("ar"),
                        common.getString("isNew"),
                        page.getString("shop_id"),
                        1L,
                        0L,
                        0L,
                        0L,
                        0L,
                        jsonObject.getLong("ts"));
            }
        });
        SingleOutputStreamOperator<TrafficPageViewBean> ujDS1 = (SingleOutputStreamOperator<TrafficPageViewBean>) ujDS.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject common = jsonObject.getJSONObject("common");
                JSONObject page = jsonObject.getJSONObject("page");
                return new TrafficPageViewBean(null,
                        null,
                        common.getString("vc"),
                        common.getString("ch"),
                        common.getString("ar"),
                        common.getString("isNew"),
                        page.getString("shop_id"),
                        0L,
                        0L,
                        0L,
                        0L,
                        1L,
                        jsonObject.getLong("ts"));
            }
        });
        SingleOutputStreamOperator<TrafficPageViewBean> pageDS1 = (SingleOutputStreamOperator<TrafficPageViewBean>) pageDS.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject common = jsonObject.getJSONObject("common");
                JSONObject page = jsonObject.getJSONObject("page");
                Long sv=0L;
                if(page.getString("last_page_id")==null){
                    sv=1L;
                }
                return new TrafficPageViewBean(null,
                        null,
                        common.getString("vc"),
                        common.getString("ch"),
                        common.getString("ar"),
                        common.getString("isNew"),
                        page.getString("shop_id"),
                        0L,
                        sv,
                        1L,
                        page.getLong("during_time"),
                        0L,
                        jsonObject.getLong("ts"));
            }
        });
        DataStream<TrafficPageViewBean> union = uvDS1.union(ujDS1, pageDS1);
        union.print("union");
        SingleOutputStreamOperator<TrafficPageViewBean> swx = union.assignTimestampsAndWatermarks(WatermarkStrategy.<TrafficPageViewBean>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean>() {
            @Override
            public long extractTimestamp(TrafficPageViewBean element, long recordTimestamp) {
                return element.getTs();
            }
        }));
        WindowedStream<TrafficPageViewBean, Tuple5<String, String, String, String,String>, TimeWindow> windowedStream = swx.keyBy(new KeySelector<TrafficPageViewBean, Tuple5<String, String, String, String,String>>() {
            @Override
            public Tuple5<String, String, String, String,String> getKey(TrafficPageViewBean value) throws Exception {
                return Tuple5.of(value.getVc(), value.getCh(), value.getAr(), value.getIsNew(), value.getShopId());
            }
        }).window(TumblingEventTimeWindows.of(Time.seconds(10))).allowedLateness(Time.seconds(10L));

        SingleOutputStreamOperator<TrafficPageViewBean> reduce = windowedStream.reduce(new ReduceFunction<TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean reduce(TrafficPageViewBean value1, TrafficPageViewBean value2) throws Exception {
                value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                value1.setUjCt(value1.getUjCt() + value2.getUjCt());
                value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                return value1;
            }
        }, new WindowFunction<TrafficPageViewBean, TrafficPageViewBean,
                Tuple5<String, String, String, String,String>, TimeWindow>() {
            @Override
            public void apply(Tuple5<String, String, String, String,String> stringStringStringStringTuple4,
                              TimeWindow window, Iterable<TrafficPageViewBean> input,
                              Collector<TrafficPageViewBean> out) throws Exception {
                TrafficPageViewBean next = input.iterator().next();
                next.setStt(new Timestamp(window.getStart()));
                next.setEdt(new Timestamp(window.getEnd()));
                next.setTs(System.currentTimeMillis());
                out.collect(next);
            }
        });
        reduce.print(">>>>>>>>");
        reduce.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_traffic_vc_ch_ar_is_new_shop_page_view_window values(?,?,?,?,?,?,?,?,?,?,?,?,?)"));
        env.execute("DwsTrafficVcChArIsNewPageViewWindow");

    }

}
